package com.xrui.hbase;

import com.google.common.base.Objects;
import com.google.common.base.Preconditions;

import com.xrui.hbase.exception.TableNotFoundException;
import com.xrui.hbase.impl.HBaseBufferedWriter;
import com.xrui.hbase.impl.HBaseTableReader;
import com.xrui.hbase.impl.HBaseTableWriter;
import com.xrui.hbase.schema.TableLayout;
import com.xrui.hbase.util.DebugResourceTracker;
import com.xrui.hbase.util.ResourceUtils;
import org.apache.hadoop.hbase.client.HConnection;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;

/**
 * <p>A HBaseTable that exposes the underlying HBase implementation.</p>
 * <p>
 * <p>Within the internal code, we use this class so that we have
 * access to the HTable interface.  Methods that HBase clients should
 * have access to should be added to com.tr.ts.HBaseTable.</p>
 */
public final class HBaseTable implements Table {
    //region static
    private static final Logger LOG = LoggerFactory.getLogger(HBaseTable.class);
    //endregion

    //region Variables
    /**
     * Tracks the state of this hbase table.
     */
    private final AtomicReference<State> mState = new AtomicReference<>(State.UNINITIALIZED);
    /**
     * The hbase instance this table belongs to.
     */
    private final HBase mHBase;
    /**
     * The name of this table.
     */
    private final String mName;
    /**
     * URI of this table.
     */
    private final HBaseURI mTableURI;
    /**
     * Retain counter. When decreased to 0, the HBase HBaseTable may be closed and disposed of.
     */
    private final AtomicInteger mRetainCount = new AtomicInteger(0);
    /**
     * HConnection used for creating lightweight tables. Should be closed by HBase.
     */
    private final HConnection mHConnection;

    /**/
    private final TableLayout mTableLayout;

    /**
     * Construct an opened HBase table stored in HBase.
     *
     * @param hbase       The HBase instance.
     * @param name        The name of the user-space table to open.

     * @param hConnection A factory that creates HTable objects.
     * @throws IOException            On an HBase error.
     * @throws TableNotFoundException if the table does not exist.
     */
    HBaseTable(
        HBase hbase,
        String name,
        HConnection hConnection
    ) throws IOException {
        mHBase = hbase;
        mHBase.retain();

        mName = name;
        mHConnection = hConnection;
        mTableURI = HBaseURI.newBuilder(mHBase.getURI()).withTableName(mName).build();

        //if (!mHBase.getTableNames().contains(mName)) {
        //    closeResources();
        //    throw new TableNotFoundException(mTableURI);
        //}

        //mWriterFactory = new HBaseWriterFactory(this);
        //mReaderFactory = new HBaseReaderFactory(this);

        mTableLayout = mHBase.getTableLayout(mTableURI);

        // DatabaseTable is now open and must be released properly:
        mRetainCount.set(1);

        final State oldState = mState.getAndSet(State.OPEN);
        Preconditions.checkState(oldState == State.UNINITIALIZED,
            "Cannot open HBaseTable instance in state %s.", oldState);
        DebugResourceTracker.get().registerResource(this);
    }
    //endregion

    //region default construct: Only can be instance by HBase

    /**
     * Opens a new connection to the HBase table backing this HBase table.
     * <p>
     * <p> The caller is responsible for properly closing the connection afterwards. </p>
     * <p>
     * Note: this does not necessarily create a new HTable instance, but may instead return
     * an already existing HTable instance from a pool managed by this HBaseHBaseTable.
     * Closing a pooled HTable instance internally moves the HTable instance back into the pool.
     * </p>
     *
     * @return A new HTable associated with this HBaseTable.
     * @throws IOException in case of an error.
     */
    public HTableInterface openHTableConnection() throws IOException {
        final State state = mState.get();
        Preconditions.checkState(state == State.OPEN,
            "Cannot open an HTable connection for a HBaseTable in state %s.", state);
        return mHConnection.getTable(mName);
    }

    //endregion

    //region APIs

    @Override
    public HBase getHBase() {
        return mHBase;
    }

    @Override
    public String getName() {
        return mName;
    }

    @Override
    public HBaseURI getURI() {
        return mTableURI;
    }

    @Override
    public TableLayout getLayout() {
        return mTableLayout;
    }

    @Override
    public EntityId getEntityId(Object... rowKey) throws IOException {
        return HBaseEntityId.create(mTableLayout, rowKey);
    }

    @Override
    public EntityId getEntityIdFromRowKey(byte[] rowKey) throws IOException {
        return HBaseEntityId.create(rowKey, mTableLayout);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public TableReader openTableReader() {
        final State state = mState.get();
        Preconditions.checkState(state == State.OPEN,
            "Cannot open a table reader on a Table in state %s.", state);
        try {
            return HBaseTableReader.create(this);
        } catch (IOException ioe) {
            throw new com.xrui.hbase.exception.IOException(ioe);
        }
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public TableWriter openTableWriter() {
        final State state = mState.get();
        Preconditions.checkState(state == State.OPEN,
            "Cannot open a table writer on a Table in state %s.", state);
        try {
            return new HBaseTableWriter(this);
        } catch (IOException ioe) {
            throw new com.xrui.hbase.exception.IOException(ioe);
        }
    }

    @Override
    public BufferedWriter openBufferedWriter() {
        final State state = mState.get();
        Preconditions.checkState(state == State.OPEN,
            "Cannot open a table buffered writer on a Table in state %s.", state);
        try {
            return new HBaseBufferedWriter(this);
        } catch (IOException ioe) {
            throw new com.xrui.hbase.exception.IOException(ioe);
        }
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public HBaseTable retain() {
        final int counter = mRetainCount.getAndIncrement();
        Preconditions.checkState(counter >= 1,
            "Cannot retain a closed HBaseTable %s: retain counter was %s.", mTableURI, counter);
        return this;
    }
    //@Override
    //public ReaderFactory getReaderFactory() throws IOException {
    //    final State state = mState.get();
    //    Preconditions.checkState(state == State.OPEN,
    //        "Cannot create the reader factory for a Table in state %s.", state);
    //    return mReaderFactory;
    //}

    //@Override
    //public WriterFactory getWriterFactory() throws IOException {
    //    final State state = mState.get();
    //    Preconditions.checkState(state == State.OPEN,
    //        "Cannot create the writer factory for a Table in state %s.", state);
    //    return mWriterFactory;
    //}

    /**
     * {@inheritDoc}
     */
    @Override
    public void release() throws IOException {
        final int counter = mRetainCount.decrementAndGet();
        Preconditions.checkState(counter >= 0,
            "Cannot release closed HBaseTable %s: retain counter is now %s.", mTableURI, counter);
        if (counter == 0) {
            closeResources();
        }
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public boolean equals(Object obj) {
        if (null == obj) {
            return false;
        }
        if (obj == this) {
            return true;
        }
        if (!getClass().equals(obj.getClass())) {
            return false;
        }
        final HBaseTable other = (HBaseTable) obj;

        // Equal if the two tables have the same URI:
        return mTableURI.equals(other.getURI());
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public int hashCode() {
        return mTableURI.hashCode();
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public String toString() {
        return Objects.toStringHelper(HBaseTable.class)
            .add("id", System.identityHashCode(this))
            .add("uri", mTableURI)
            .add("retain_counter", mRetainCount.get())
            .add("state", mState.get())
            .toString();
    }

    /**
     * Releases the resources used by this table.
     *
     * @throws IOException on I/O error.
     */
    private void closeResources() throws IOException {
        final State oldState = mState.getAndSet(State.CLOSED);
        Preconditions.checkState(oldState == State.OPEN || oldState == State.UNINITIALIZED,
            "Cannot close HBaseTable instance %s in state %s.", this, oldState);
        LOG.debug("Closing HBase Table '{}'.", this);

        ResourceUtils.releaseOrLog(mHBase);

        if (oldState != State.UNINITIALIZED) {
            DebugResourceTracker.get().unregisterResource(this);
        }

        LOG.debug("HBase Table '{}' closed.", mTableURI);
    }
    //endregion

    //region private method

    /**
     * States of a HBase table instance.
     */
    private enum State {
        /**
         * Initialization begun but not completed.  Retain counter and DebugResourceTracker counters
         * have not been incremented yet.
         */
        UNINITIALIZED,
        /**
         * Finished initialization.  Both retain counters and DebugResourceTracker counters have been
         * incremented.  Resources are successfully opened and this HBaseHBaseTable's methods may be
         * used.
         */
        OPEN,
        /**
         * Closed.  Other methods are no longer supported.  Resources and connections have been closed.
         */
        CLOSED
    }
    //endregion


}
